package com.ry.flink.jobx;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * 1. 创建临时视图
 * */
public class Job1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tablEnv = StreamTableEnvironment.create(env);
        DataStreamSource<Tuple2<String, String>> dataStream = env.fromElements(Tuple2.of("1L", "Alice"), Tuple2.of("2L", "Bob"));
        Table table = tablEnv.fromDataStream(dataStream);

        tablEnv.createTemporaryView("MyView",table);
        /**
         * 默认的Schema信息：
         *  (
         *        `f0` STRING,
         *        `f1` STRING
         *  )
         * */
//        tablEnv.from("MyView").printSchema();

        /**
         * 指定Schema信息（方式一）：
         *         (
         *                `f0` BIGINT,
         *                `f1` STRING
         *          )
         * */
        tablEnv.createTemporaryView(
                "MyView2",
                dataStream,
                Schema.newBuilder()
                    .column("f0","BIGINT")
                    .column("f1","STRING")
                    .build());
//        tablEnv.from("MyView2").printSchema();

        /**
         * 指定Schema信息（方式二）：
         *   `id` STRING,
         *   `name` STRING
         *
         * */
        tablEnv.createTemporaryView("MyView3",tablEnv.fromDataStream(dataStream).as("id","name"));
        tablEnv.from("MyView3").printSchema();
        env.execute("Job1");
    }
}
